package mapReduce.homework.q6_orderInfoTop10;

import mapReduce.util.JobSubmit;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class GetMenuSailCountApp {

    public static class oitMapper extends Mapper<LongWritable,Text,OrderInfoWritable, IntWritable>{

        OrderInfoWritable outkey = new OrderInfoWritable();
        IntWritable outValue = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            outkey.setYear(fields[0]);
            outkey.setOrderId(fields[1]);
            outkey.setPartner(fields[2]);
            outkey.setmenuId(fields[3]);
            outkey.setMoney(Double.parseDouble(fields[4]));

            context.write(outkey,outValue);
        }
    }

    public static class oitReducer extends Reducer<OrderInfoWritable,IntWritable,OrderInfoWritable,IntWritable>{

        List<Map<OrderInfoWritable,Integer>> menuList = new ArrayList<Map<OrderInfoWritable,Integer>>();

        @Override
        protected void reduce(OrderInfoWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

            //初始化每年每个合作方的订单数量和
            int sum = 0;


            for (IntWritable value : values) {
                sum += value.get();
            }

            context.write(key,new IntWritable(sum));

        }
    }

    public static void main(String[] args) throws Exception {
        JobSubmit.submitBaseJob( GetMenuSailCountApp.class,args,
                "group", GetMenuSailCountGroupComparator.class);
    }

}
